
#include "spdk/stdinc.h"

#include "spdk/barrier.h"
#include "spdk/bdev.h"
#include "spdk/bdev_module.h"
#include "spdk/env.h"
#include "spdk/event.h"
#include "spdk/likely.h"
#include "spdk/thread.h"
#include "spdk/json.h"
#include "spdk/util.h"
#include "spdk/string.h"
#include "spdk/log.h"

#include "./bdev_zcell.h"
#include "../liboss.h"
#include "../spdk_ipc_config.h"

#include <pthread.h>

#define ZCELL_MAX (1024)
#define MAX_EVENTS_PER_POLL (128)

#define BLOCK_SIZE (4ul<<10)
#define BLOCK_SIZE_SHIFT (12)
#define STRIPE_UNIT (4ul<<20)
#define STRIPE_UNIT_SHIFT (22)

static TAILQ_HEAD( , zcell_disk) g_zcell_disk_head;
static uint64_t g_disk_offset;

//Vdisk ==> Object map policy
// static uint64_t GetOidStart(size_t GiB , uint64_t stripes) {
//     uint64_t objs = GiB  << (30 - STRIPE_UNIT_SHIFT);
//     objs /= stripes;
//     uint64_t oid = oid_max;
//     oid_max += objs;
//     return oid; 
// }

/**
 * 
 * vdisk 按照 RAID0 方式以 4M 
 * 为粒度条带化到所有 zcell 上。
 * 
 * liboss_poller
 * |Polling
 * zcell0_session
 * zcell1_session
 * zcell2_session
 * zcell3_session
 * 
 */
// struct zcell_iochannel {
//     struct io_channel * ch;
//     TAILQ_ENTRY(zcell_iochannel) link;
// };


struct zcell_channel_group {
    void *liboss_ctx;
    uint32_t refcount;
    uint32_t mylcore;
    struct spdk_poller	*completion_poller;
    //已经连接的 local zcell 的 io_channels
    uint32_t zcell_nr;
    struct io_channel *zioch_list[128];
};




static struct zcell_channel_group g_zgrps[1024];

static struct zcell_ipc_config_t *g_zcfg;

static pthread_spinlock_t g_sl;
static int g_zgrp_living;
static uint32_t g_zcell_serve_rate_allocated;

struct bdev_zcell_channel_group {
    struct zcell_channel_group *lzgrp;
};

struct zcell_ioctx {
    int zcell_op_id; //  
};


static int get_qos_tag_id() {
    static i = 1 ;
    return i++;
}

struct zcell_disk {
    struct spdk_bdev disk;
    uint64_t disk_offset; //GiB
    
    int32_t qos_iops;
    uint32_t qos_tag_id;

    TAILQ_ENTRY(zcell_disk) link;
};

struct zcell_disk_io_channel {
    uint64_t io_inflight;
    struct zcell_channel_group *zgrp;
};


static struct spdk_io_channel *
bdev_zcell_get_io_channel(void *ctx)
{
	struct zcell_disk *zdisk = ctx;
    // struct spdk_thread *st = spdk_get_thread();
    // const char *stname = spdk_thread_get_name(st);
    // SPDK_NOTICELOG("thname = %s  , get_io_channel \n" , stname );
	return spdk_get_io_channel(zdisk);
}

static bool
bdev_zcell_io_type_supported(void *ctx, enum spdk_bdev_io_type io_type)
{
	switch (io_type) {
	case SPDK_BDEV_IO_TYPE_READ:
	case SPDK_BDEV_IO_TYPE_WRITE:
	case SPDK_BDEV_IO_TYPE_FLUSH:
	case SPDK_BDEV_IO_TYPE_RESET:
		return true;
	default:
		return false;
	}
}

static int
bdev_zcell_destruct(void *ctx)
{
	struct zcell_disk *disk = ctx;
	TAILQ_REMOVE(&g_zcell_disk_head, disk, link);
	spdk_io_device_unregister(disk, NULL);
    free(disk);
	return 0;
}


static int
bdev_zcell_dump_info_json(void *ctx, struct spdk_json_write_ctx *w)
{
	struct zcell_disk *fdisk = ctx;
	spdk_json_write_named_object_begin(w, "zcell");
	spdk_json_write_named_string(w, "diskname", fdisk->disk.name);
	spdk_json_write_object_end(w);
	return 0;
}

static void
bdev_zcell_write_json_config(struct spdk_bdev *bdev, struct spdk_json_write_ctx *w)
{
	spdk_json_write_object_begin(w);

	spdk_json_write_object_end(w);
}


static void
bdev_zcell_get_buf_cb(struct spdk_io_channel *ch, struct spdk_bdev_io *bdev_io,
		    bool success)
{
	if (!success) {
		spdk_bdev_io_complete(bdev_io, SPDK_BDEV_IO_STATUS_FAILED);
		return;
	}
	struct zcell_disk_io_channel *zch = spdk_io_channel_get_ctx(ch);
    struct zcell_channel_group *zgrp = zch->zgrp;
    struct zcell_disk *zdisk = (struct zcell_disk *)(bdev_io->bdev->ctxt);


    struct zcell_ioctx *zio = (struct zcell_ioctx *)(bdev_io->driver_ctx);
    struct iovec *iov = bdev_io->u.bdev.iovs;
    uint32_t iovcnt = bdev_io->u.bdev.iovcnt;
    uint32_t len = bdev_io->u.bdev.num_blocks * bdev_io->bdev->blocklen;
    uint64_t offset = bdev_io->u.bdev.offset_blocks * bdev_io->bdev->blocklen;

    uint64_t stripe_id = (offset >> STRIPE_UNIT_SHIFT) % zgrp->zcell_nr ;  
    uint64_t obj_id =  ((zdisk->disk_offset << 8)  + (offset >> STRIPE_UNIT_SHIFT)) / zgrp->zcell_nr;


    uint64_t obj_offset = (offset % STRIPE_UNIT);
    uint64_t obj_len = len;

    if(iovcnt > 1) {
        SPDK_ERRLOG("iovcnt > 1\n");
        // assert(iovcnt == 1);
        // return -1;
        spdk_bdev_io_complete(bdev_io, SPDK_BDEV_IO_STATUS_FAILED);
        return;
    }

    if(iov[0].iov_base == NULL) {
        SPDK_ERRLOG("iov[0].iov_base == NULL\n");
        spdk_bdev_io_complete(bdev_io, SPDK_BDEV_IO_STATUS_FAILED);
        return;
    }

    if(bdev_io->type == SPDK_BDEV_IO_TYPE_READ) {
        zio->zcell_op_id = io_read2(zgrp->zioch_list[stripe_id] , iov->iov_base ,
            obj_id , obj_offset , obj_len ); 
    } else {
        zio->zcell_op_id = io_write(zgrp->zioch_list[stripe_id] ,
            obj_id , iov->iov_base , obj_offset , obj_len );
    }

    if(zio->zcell_op_id < 0) {
        spdk_bdev_io_complete(bdev_io, SPDK_BDEV_IO_STATUS_FAILED);
        return;
    }

    op_set_userdata(zgrp->zioch_list[stripe_id] , 
        zio->zcell_op_id, (uint64_t)(bdev_io));

    op_set_qos_tag(zgrp->zioch_list[stripe_id] , 
        zio->zcell_op_id, zdisk->qos_tag_id);
    
    int op_ids[] = { zio->zcell_op_id };

    int rc = io_submit_to_channel(zgrp->zioch_list[stripe_id] , op_ids , 1);
    
    if(rc) {
        spdk_bdev_io_complete(bdev_io, SPDK_BDEV_IO_STATUS_FAILED);
        return;
    }  
    return ;

}

static int _bdev_zcell_submit_request(struct spdk_io_channel *ch, struct spdk_bdev_io *bdev_io)
{

    int rc;
    (void)rc;
	switch (bdev_io->type) {
	case SPDK_BDEV_IO_TYPE_READ:
	case SPDK_BDEV_IO_TYPE_WRITE: {
        spdk_bdev_io_get_buf(bdev_io, bdev_zcell_get_buf_cb,
				     bdev_io->u.bdev.num_blocks * bdev_io->bdev->blocklen);
		return 0;
		// spdk_bdev_io_get_buf(bdev_io, bdev_aio_get_buf_cb,
		// bdev_io->u.bdev.num_blocks * bdev_io->bdev->blocklen);

    }

	case SPDK_BDEV_IO_TYPE_FLUSH:
	case SPDK_BDEV_IO_TYPE_RESET: {
        spdk_bdev_io_complete(bdev_io , SPDK_BDEV_IO_STATUS_SUCCESS);
		return 0;
    }

	default:
		return -1;
	}
}

static void bdev_zcell_submit_request(struct spdk_io_channel *ch, struct spdk_bdev_io *bdev_io)
{
	if (_bdev_zcell_submit_request(ch, bdev_io) < 0) {
		spdk_bdev_io_complete(bdev_io, SPDK_BDEV_IO_STATUS_FAILED);
	}
}


static const struct spdk_bdev_fn_table zcell_disk_fn_table = {
	.destruct		= bdev_zcell_destruct,
	.submit_request		= bdev_zcell_submit_request,
	.io_type_supported	= bdev_zcell_io_type_supported,
	.get_io_channel		= bdev_zcell_get_io_channel,
	.dump_info_json		= bdev_zcell_dump_info_json,
	.write_config_json	= bdev_zcell_write_json_config,
};


/**
 * 
 * 
 * 
 */
static int bdev_zcell_initialize(void);
static void bdev_zcell_fini(void);
static int bdev_zcell_get_ctx_size(void);

static struct spdk_bdev_module zcell_if = {
	.name		= "zcell",
	.module_init	= bdev_zcell_initialize,
	.module_fini	= bdev_zcell_fini,
	.get_ctx_size	= bdev_zcell_get_ctx_size,
};

SPDK_BDEV_MODULE_REGISTER(zcell, &zcell_if)



static int
bdev_zcell_create_cb(void *io_device, void *ctx_buf)
{
    struct zcell_disk_io_channel *ch = ctx_buf;
    struct zcell_disk *zdisk = io_device;
    (void)zdisk;

    uint32_t c = spdk_env_get_current_core();
    if(c == SPDK_ENV_LCORE_ID_ANY) {
        SPDK_ERRLOG("get_io_channel的调用者不是一个绑了核的spdk_thread\n");
        struct spdk_thread *st = spdk_get_thread();
        const char *stname = spdk_thread_get_name(st);
        SPDK_NOTICELOG("zcell_group_iochannel = %p , thname = %s  \n" , ch->zgrp , stname );  
        assert(false);
    }
    ch->zgrp = &(g_zgrps[c]);
    // ch->bzgrp = spdk_io_channel_get_ctx(spdk_get_io_channel(&zcell_if)); 
    return 0;
}

static void
bdev_zcell_destroy_cb(void *io_device, void *ctx_buf)
{
    // struct zcell_disk_io_channel *ch = ctx_buf;

    // struct spdk_thread *st = spdk_get_thread();
    // const char *stname = spdk_thread_get_name(st);
    // SPDK_NOTICELOG("zcell_group_iochannel = %p , thname = %s  \n" , ch->zgrp , stname );
	
}

int create_zcell_disk(const char *name , uint32_t size_GiB , int32_t iops)
{
    struct zcell_disk *zdisk;
    zdisk = calloc(1, sizeof(*zdisk));
	if (!zdisk) {
		SPDK_ERRLOG("Unable to allocate enough memory for zcell backend\n");
		return -ENOMEM;
	}
    zdisk->disk_offset = g_disk_offset;
    g_disk_offset += (size_GiB);

    if(iops == -1 ) {
        //BE 租户
        zdisk->qos_tag_id = 0;
        zdisk->qos_iops = -1;
    
    } else {

        //LC 租户
        
        zdisk->qos_iops = iops;
        zdisk->qos_tag_id = get_qos_tag_id(); 
        if(zdisk->qos_tag_id < QOS_TARGET_MAX) {
            if(g_zcell_serve_rate_allocated - iops >= 0 ) {
                volatile struct zcell_qos_tag_t *zqt = &g_zcfg->qos_tags[zdisk->qos_tag_id];
                zqt->iops = iops;
                zqt->avg_latency = 0;
                volatile int *a= &g_zcfg->qos_target_state[zdisk->qos_tag_id];
                *a = 1;
                spdk_mb();

                g_zcell_serve_rate_allocated -= iops;
            } else {
                SPDK_ERRLOG("SLO 超过了最大服务能力：%u K  , 已分配：%u K，SLO ：%u K\n" , 
                g_zcfg->zcell_max_serve_rate / 1000 , 
                g_zcell_serve_rate_allocated / 1000 ,
                iops / 1000 );
                return -1;         
            }
        }
        else {
            SPDK_ERRLOG("超过了租户数量限制：%d\n" , QOS_TARGET_MAX );
            return -1;
        }

    }





    zdisk->disk.module = &zcell_if;
    zdisk->disk.blocklen = 0x1000;
    zdisk->disk.blockcnt = ((uint64_t)size_GiB) << 18;
    zdisk->disk.ctxt = zdisk;
    zdisk->disk.write_unit_size = 1;
    zdisk->disk.product_name = "Zcell Disk";
    zdisk->disk.fn_table = &zcell_disk_fn_table;
    zdisk->disk.split_on_optimal_io_boundary = true;
    zdisk->disk.name = strdup(name);
    zdisk->disk.optimal_io_boundary = STRIPE_UNIT >> 12 ;

    spdk_io_device_register(zdisk, bdev_zcell_create_cb, bdev_zcell_destroy_cb,
				sizeof(struct zcell_disk_io_channel),
				zdisk->disk.name);

    int rc = spdk_bdev_register(&zdisk->disk);
    (void)rc;


    TAILQ_INSERT_TAIL(&g_zcell_disk_head , zdisk , link);
    
    
    volatile int *a = &g_zcfg->qos_target_state[zdisk->qos_tag_id];
    *a = 1;
    spdk_mb();

    //等 1 s... 必须等所有 zcell 都接收完成
    spdk_delay_us( 1 * 1000ul * 1000ul );

    if(iops == -1 ) {
        SPDK_NOTICELOG("Create BE ZDisk:[%s] done\n" , name );       
    } else { 
        SPDK_NOTICELOG("Create LC ZDisk:[%s] done. [Qos Tag]:< id=%d , iops=%lfK , latency=%dus >\n" , 
            name ,
            zdisk->qos_tag_id ,
            (double) zdisk->qos_iops / 1000.0 , -1 );
    }
    
    return 0;
}

struct delete_zcell_bdev_ctx {
	delete_zcell_bdev_complete cb_fn;
	void *cb_arg;
};

static void
zcell_bdev_unregister_cb(void *arg, int bdeverrno)
{
	struct delete_zcell_bdev_ctx *ctx = arg;
	ctx->cb_fn(ctx->cb_arg, bdeverrno);
	free(ctx);
}

void delete_zcell_disk(struct spdk_bdev *bdev, spdk_bdev_unregister_cb cb_fn,
			  void *cb_arg)
{
	struct delete_zcell_bdev_ctx *ctx;

    struct zcell_disk *zd = (struct zcell_disk *)bdev->ctxt;
    int id = zd->qos_tag_id;

    volatile int *a = &g_zcfg->qos_target_state[id];
    *a = 0;
    spdk_mb();


	if (!bdev || bdev->module != &zcell_if) {
		cb_fn(cb_arg, -ENODEV);
		return;
	}

	ctx = calloc(1, sizeof(*ctx));
	if (ctx == NULL) {
		cb_fn(cb_arg, -ENOMEM);
		return;
	}

	ctx->cb_fn = cb_fn;
	ctx->cb_arg = cb_arg;
	spdk_bdev_unregister(bdev, zcell_bdev_unregister_cb, ctx);

}

static int 
bdev_zcell_group_poll( void * grp_)
{
    struct zcell_channel_group *grp = grp_;
    uint32_t i;
    uint32_t count = 0;

    for ( i = 0 ; i < grp->zcell_nr ; ++i) {
        struct io_channel *ch = grp->zioch_list[i];
        int cpls[MAX_EVENTS_PER_POLL];
        int n = io_poll_channel(ch , cpls , 0 , MAX_EVENTS_PER_POLL);
        assert(n >= 0);
        if( n > 0 ) {
            int j;
            for ( j = 0 ; j < n ; ++j) {
                uint64_t bio_ = op_get_userdata(ch, cpls[j]);
                int errcode ;
                void *databuf;
                uint32_t datalen;
                int rc = op_claim_result(ch , cpls[j] , &errcode , NULL, &databuf, &datalen);
                if(rc) {
                    assert(false);
                }
                op_destory(ch , cpls[j]);
                if(errcode) {
                    errcode = SPDK_BDEV_IO_STATUS_FAILED;
                } else {
                    errcode = SPDK_BDEV_IO_STATUS_SUCCESS;
                }
                struct spdk_bdev_io *bio = (void *)(uintptr_t)(bio_);
                spdk_bdev_io_complete(bio , errcode);
            }
            count += n;
        }

    }


    return count;
}

static void
_init_zcell_group(void *grp_ , void *foo)
{
    struct zcell_channel_group * grp = grp_;
    if(grp->refcount == 0) {
        
        // SPDK_WARNLOG("Current core=%d\n" , spdk_env_get_current_core());
        // if(spdk_env_get_current_core() == SPDK_ENV_LCORE_ID_ANY) {

        // };
        int rc = tls_io_ctx_init(grp->mylcore , &(grp->liboss_ctx)); // 初始化当前 lcore 的 msgr     
        if(rc) {
            SPDK_ERRLOG("liboss ctx tls init failed\n");
            return;
        }
        grp->zcell_nr =  get_io_channel_with_local_all(512 , grp->zioch_list);
        // grp->zcell_nr = 1;
        // grp->zioch_list[0] = get_io_channel_with_local(0 , 512);
        if(!grp->zcell_nr) {
            SPDK_ERRLOG("liboss get channel with all failed \n");
            return;
        }
        grp->completion_poller = spdk_poller_register(bdev_zcell_group_poll , grp , 0 );
        if(!grp->completion_poller) {
            SPDK_ERRLOG("spdk_poller_register  failed \n");
            return;
        }
    }

    pthread_spin_lock(&g_sl);
    g_zgrp_living++;
    pthread_spin_unlock(&g_sl);

    return;
}

static void
_fini_zcell_group(void *grp_ ,  void *foo)
{
	struct zcell_channel_group *grp = grp_;

    spdk_poller_unregister(&grp->completion_poller);
    uint64_t i;
    for (i = 0 ; i < grp->zcell_nr ; ++i) {
        liboss_put_io_channel(grp->zioch_list[i]);
    }   
    tls_io_ctx_fini(grp->liboss_ctx); //销毁当前线程的 msgr    

    pthread_spin_lock(&g_sl);
    g_zgrp_living--;
    pthread_spin_unlock(&g_sl);

    return;
}

static int
bdev_zcell_group_create_cb(void *io_device, void *ctx_buf)
{
	struct bdev_zcell_channel_group *bzg = ctx_buf;
    uint32_t c = spdk_env_get_current_core();
    if(c == SPDK_ENV_LCORE_ID_ANY) {
        SPDK_ERRLOG("get_io_channel的调用者不是一个绑了核的spdk_thread\n");
        return -1;
    }
    bzg->lzgrp = &g_zgrps[c];
    bzg->lzgrp->refcount++;
    return 0;
}

static void
bdev_zcell_group_destroy_cb(void *io_device, void *ctx_buf)
{
	struct bdev_zcell_channel_group *bzg = ctx_buf;
    // _fini_zcell_group(ctx_buf, NULL);
    bzg->lzgrp->refcount--;
}


static int bdev_zcell_get_ctx_size(void)
{
    return sizeof(struct zcell_ioctx);
}


enum {
    INIT_ZCELL_GROUP = 1,
    DESTORY_ZCELL_GROUP = 2,
};
static void for_each_reactor ( int opcode )
{
    spdk_event_fn fn; 
    if(opcode == INIT_ZCELL_GROUP) {
        fn = _init_zcell_group;
    } else if (opcode == DESTORY_ZCELL_GROUP) {
        fn = _fini_zcell_group;
    } else {
        fn = NULL;
    }

    uint32_t core_count = spdk_env_get_core_count();
    // SPDK_NOTICELOG("cores=%d\n",core_count);
    uint32_t master_core = spdk_env_get_first_core();
    // SPDK_NOTICELOG("master_core=%d\n",master_core);

    uint32_t i;
    SPDK_ENV_FOREACH_CORE(i) {
        uint32_t cur = i;
        g_zgrps[i].mylcore = i;
        if (cur != master_core) {
            struct  spdk_event *e = 
            spdk_event_allocate( i , fn , &g_zgrps[i] , NULL);
            spdk_event_call(e);
        }
    }

    uint32_t lives = 0;
    do {
        pthread_spin_lock(&g_sl);
        lives = g_zgrp_living;
        pthread_spin_unlock(&g_sl);
    } while ( (opcode == INIT_ZCELL_GROUP && lives != core_count - 1) 
        || (opcode == DESTORY_ZCELL_GROUP && lives != 1));
    // SPDK_NOTICELOG("waiting for master core fn \n");

    fn( &g_zgrps[master_core] , NULL );

    return ;
}



static int
bdev_zcell_initialize(void)
{
    TAILQ_INIT(&g_zcell_disk_head);

    g_zcfg = spdk_memzone_lookup(ZCELL_IPC_CONFIG_NAME);
    if(!g_zcfg) {
        SPDK_ERRLOG("fuck\n");
        return -1;
    }

    g_zcell_serve_rate_allocated = g_zcfg->zcell_max_serve_rate;

    pthread_spin_init(&g_sl , 0);

    //
    for_each_reactor(INIT_ZCELL_GROUP);

    // spdk_io_device_register(&zcell_if, bdev_zcell_group_create_cb, bdev_zcell_group_destroy_cb,
	// 			sizeof(struct bdev_zcell_channel_group), "zcell_module");

	return 0;
}

static void
bdev_zcell_fini(void)
{

    // spdk_io_device_unregister(&zcell_if, NULL);

    for_each_reactor(DESTORY_ZCELL_GROUP);
    
    pthread_spin_destroy(&g_sl);
}


// SPDK_LOG_REGISTER_COMPONENT("zcell", SPDK_LOG_ZCELL)